#include "../config.h"

#include <glibmm/i18n.h>

#include <string.h>
#include <boost/algorithm/string.hpp>

#include "application.h"
#include "log.h"
#include "loader.h"
#include "strfuncs.h"

#ifdef DEBUG
# ifdef HAVE_BOOST_ASIO
#  warning using boost::asio
# else
#  warning using asio
# endif
#endif

namespace gpyp {
namespace loader {

//========================
// Dispatcher
// Glib::Dispatcherはオブジェクトを破棄した時に呼び出しをキャンセルしてくれなくて
// 色々と面倒なので自前で実装する
gboolean Dispatcher::on_idle(Data *data) {
  Dispatcher *self = data->self;
  bool match = false;
  g_mutex_lock(self->mutex_);
  g_source_remove(data->id);
  data->id = 0;
  for(boost::ptr_list<Data>::iterator iter = self->data_.begin();
      iter != self->data_.end(); ++iter) {
    if (&*iter == data) {
      self->data_.erase(iter);
      match = true;
      break;
    }
  }
  g_mutex_unlock(self->mutex_);
  if (match && !self->slot_.empty()) self->slot_();
  return FALSE;
}

void Dispatcher::emit() {
  g_mutex_lock(mutex_);
  Data *data = new Data(*this);
  data_.push_back(data);
  data->id = g_idle_add((GSourceFunc)&Dispatcher::on_idle, data);
  g_mutex_unlock(mutex_);
}

void Dispatcher::stop() {
  g_mutex_lock(mutex_);
  for(boost::ptr_list<Data>::iterator iter = data_.begin(); iter != data_.end(); ++iter) {
    g_source_remove(iter->id);
  }
  data_.clear();
  g_mutex_unlock(mutex_);
}

Dispatcher::Dispatcher() :
  mutex_(g_mutex_new())
{
}
Dispatcher::~Dispatcher() {
  stop();

  GMutex *m = mutex_;
  g_mutex_lock(m);
  mutex_ = NULL;
  g_mutex_unlock(m);
  g_mutex_free(m);
}

//========================
// Connection
const int Connection::TIMEOUT_DEFAULT = -1;

void Connection::on_join() {
  stop_timeout_timer();
  if (loading_) {
    loading_ = false;
    if (!slot_joined_.empty()) slot_joined_();
  }
}
void Connection::on_load_thread() {
  on_load();
  join();
}
void Connection::on_cancel() {
  dispatcher_complete_.stop();
  dispatcher_error_   .stop();
}

void Connection::on_destroy() {
  dispatcher_complete_.disconnect();
  dispatcher_error_   .disconnect();
  slot_joined_ = SlotJoined();
  if (is_loading()) {
    log::log << "Connection::on_destroy: cancelled" << log::endl;
    on_cancel();
    on_join();
  }
}

void Connection::timer_thread() {
  while(1) {
    sync_mutex_.lock();
    if (sync_timer_ == 0 ||
        sync_timer_ == -2) {
      sync_mutex_.unlock();
      break;
    } else if (sync_timer_ > 0) {
      --sync_timer_;
    }
    sync_mutex_.unlock();
    sleep(1);
  }
  if (sync_timer_ == 0) timer_on_timeout();
  join();
}

void Connection::load(bool sync) {
  if (dirty_) {
    log::err << "bug: Connection is dirty" << log::endl;
    return;
  }

  timeout_ = false;
  if ((sync_ = sync)) {
    start_timeout_timer();
    app_.thread(sigc::mem_fun(*this, &Connection::timer_thread));
    loading_ = true;
    on_load();
    stop_timeout_timer();
    if (error_->is_set()) throw error_;
  } else {
    start_timeout_timer();
    app_.thread(sigc::mem_fun(*this, &Connection::on_load_thread));
    loading_ = true;
  }
}
void Connection::cancel() {
  if (loading_ && !timeout_) {
    cancelled_ = true;
    stop_timeout_timer();
    on_cancel();

    error_->set_message(_("Cancelled"));
  }
  if (next_) next_->cancel();
}

void Connection::stop_timeout_timer(bool pause) {
  if (sync_) {
    sync_mutex_.lock();
    sync_timer_ = pause ? -1 : -2;
    sync_mutex_.unlock();
  } else {
    timeout_connection_.disconnect();
  }
}
void Connection::start_timeout_timer() {
  if (sync_) {
    sync_mutex_.lock();
    sync_timer_ = (timeout_time_ <= 0
                   ? *app_.conf().network()->timeout()
                   : timeout_time_);
    sync_mutex_.unlock();
  } else {
    timeout_connection_ =
      Glib::signal_timeout()
      .connect(sigc::mem_fun(*this, &Connection::timer_on_timeout),
               (timeout_time_ <= 0
                ? *app_.conf().network()->timeout()
                : timeout_time_) * 1000);
  }
}

bool Connection::timer_on_timeout() {
  if (loading_ && !cancelled_) {
    timeout_ = true;
    on_cancel();
    error_->set_message(_("Timeout"));
    dispatcher_error_.emit();
  }
  return false;
}

void Connection::complete(const char *data, size_t size) {
  if (!timeout_ && !cancelled_) {
    if (data && size) {
      Glib::ustring err;
      const char *charset = NULL;
      if (!request_->get_charset().empty()) {
        charset = request_->get_charset().c_str();
      }
      if (!response_->get_source().set(data, size, charset, err)) {
        error_->set_message(err);
        dispatcher_error_.emit();
        return;
      }
    }
    dispatcher_complete_.emit();
  }
}
void Connection::error(const Glib::ustring &message) {
  if (!timeout_ && !cancelled_) {
    error_->set_message(message);
    dispatcher_error_.emit();
  }
}
void Connection::join() {
  dispatcher_join_.emit();
}

Connection::Connection(Application &app,
                       RefRequest   request,
                       RefResponse  response,
                       SlotError    slot_error,
                       SlotJoined   slot_joined) :
  app_         (app),
  request_     (request),
  response_    (response),
  dirty_       (false),
  sync_        (false),
  slot_joined_ (slot_joined),
  timeout_time_(TIMEOUT_DEFAULT),
  sync_timer_  (0),
  loading_     (false),
  timeout_     (false),
  cancelled_   (false),
  error_       (Error::create())
{
  dispatcher_error_.connect(sigc::bind(slot_error, error_));
  dispatcher_join_ .connect(sigc::mem_fun(*this, &Connection::on_join));
}
Connection::~Connection() {
}

//========================
// HTTPRequest
std::ostream &HTTPRequest::write_to(std::ostream &s,
                                    const char   *ln) const {
  const URL::ParamList *params = get_url().get_params();

  switch(method_) {
  case HTTPRequest::METHOD_POST:
    s << "POST ";
    break;
  case HTTPRequest::METHOD_GET:
  default:
    s << "GET ";
    break;
  }
  s << get_url().get_path();
  if (params) s << *params;
  s << " HTTP/1.0" << ln;
  headers_.write_to(s, ln);
  return s;
}

HTTPRequest::HTTPRequest(Method method, const URL &url) :
  Request(url),
  method_(method)
{
  if (strcasecmp(url.get_scheme().c_str(), "http") != 0) {
    throw Error::create(format(_("Invalid URL scheme - %s"),
                               url.get_scheme().c_str()));
  }
}
HTTPRequest::HTTPRequest(const HTTPRequest &req) :
  Request   (req),
  method_   (req.method_),
  headers_  (req.headers_),
  post_data_(req.post_data_)
{
}

//========================
// HTTPConnection
Glib::StaticMutex HTTPConnection::_resolve_mutex = GLIBMM_STATIC_MUTEX_INIT;

void HTTPConnection::on_load() {
  using ASIO_NS::ip::tcp;
  mutex_.lock();
  io_service_  .reset(new ASIO_NS::io_service());
  request_buf_ .reset(new ASIO_NS::streambuf());
  response_buf_.reset(new ASIO_NS::streambuf());

  static const char *ret    = "\r\n";
  const std::string &u_host = request_->get_url().get_host();

  request_->set_header("Host", u_host);
  request_->set_header("User-Agent",
                       app_.conf().network()->useragent()->get_ua());
  request_->set_header("Connection", "close");

  std::ostream request_stream(request_buf_.get());
  request_->write_to(request_stream, ret) << ret << ret;

  if (request_->get_method() == HTTPRequest::METHOD_POST &&
      !request_->get_post_data().empty()) {
    request_stream.write(request_->get_post_data().c_str(),
                         request_->get_post_data().size());
  }
  mutex_.unlock();

  if (!_resolve_mutex.trylock()) {
    stop_timeout_timer(true);
    timeout_ = false;
    _resolve_mutex.lock();
    start_timeout_timer();
  }
  bool err = false;
  resolve_iterator ep_iter, ep_end;
  try {
    mutex_.lock();
    try {
      resolver_.reset(new tcp::resolver(*io_service_));
    } catch(...) {
      resolver_.reset();
      mutex_.unlock();
      throw Exception("failed to create resolver");
    }
    mutex_.unlock();

    std::string sn;
    if (request_->get_url().get_port() > 0) {
      sn = tostr(request_->get_url().get_port());
    } else {
      sn = request_->get_url().get_scheme();
    }
    if (*app_.conf().network()->disable_ipv6()) {
      ep_iter = resolver_->resolve(tcp::resolver::query(tcp::v4(), u_host, sn));
    } else {
      ep_iter = resolver_->resolve(tcp::resolver::query(u_host, sn));
    }
  } catch(const std::exception &ex) {
    error(ex.what());
    err = true;
  } catch(const Glib::Exception &ex) {
    error(ex.what());
    err = true;
  } catch(const Exception &ex) {
    error(ex.what());
    err = true;
  } catch(...) {
    error(_("Unknown error"));
    err = true;
  }
  _resolve_mutex.unlock();

  if (!err && !timeout_ && !cancelled_) {
    try {
      mutex_.lock();
      socket_.reset(new tcp::socket(*io_service_));
      mutex_.unlock();
      error_code ec = ASIO_NS::error::host_not_found;
      while(ec && ep_iter != ep_end) {
        socket_->close();
        socket_->connect(*ep_iter++, ec);
      }
      if (ec) throw Exception(ec.message());
      ASIO_NS::write(*socket_, *request_buf_, ASIO_NS::transfer_all(), ec);
      if (ec) throw Exception(ec.message());
      ASIO_NS::read_until(*socket_, *response_buf_, "\n", ec);
      if (ec) throw Exception(ec.message());

      std::istream response_stream(response_buf_.get());
      {
        std::string http_version;
        int         code;
        std::string message;
        try {
          response_stream >> http_version;
          response_stream >> code;
          std::getline(response_stream, message);
          boost::trim(message);
          if (!response_stream ||
              http_version.compare(0, 5, "HTTP/")) {
            throw Exception();
          }
          response_->set_response(http_version, code, message);
        } catch(...) {
          throw Exception(_("Invalid response"));
        }
        log::dbg() << request_->get_url().get_host() << ": "
                   << format("%i %s",
                             response_->get_code(),
                             response_->get_message().c_str())
                   << log::endl;
      }
      ASIO_NS::read(*socket_, *response_buf_, ASIO_NS::transfer_all(), ec);
      if (ec && ec != ASIO_NS::error::eof) throw Exception(ec.message());

      {
        std::string header;
        while (std::getline(response_stream, header) &&
               !header.empty() && header != "\r") {
          std::string::size_type s = header.find(':');
          if (s != std::string::npos) {
            std::string key(header, 0, s);
            std::string val(header, s + 1, std::string::npos);
            boost::trim(key);
            boost::trim(val);
            response_->set_header(key, val);
          }
        }

        size_t size = response_buf_->size();
        if (size) {
          char *data = new char[size];
          response_stream.read(data, size);
          complete(data, size);
          delete [] data;
        } else {
          complete(NULL, 0);
        }
      }
    } catch(const std::exception &ex) {
      error(ex.what());
    } catch(const Glib::Exception &ex) {
      error(ex.what());
    } catch(const Exception &ex) {
      error(ex.what());
    } catch(...) {
      error(_("Unknown error"));
    }
  }
  mutex_.lock();
  response_buf_.reset();
  request_buf_ .reset();
  socket_      .reset();
  resolver_    .reset();
  io_service_  .reset();
  mutex_.unlock();
}
void HTTPConnection::on_cancel() {
  Connection::on_cancel();
  mutex_.lock();
  if (io_service_) {
    io_service_->stop();
    if (resolver_) resolver_->cancel();
    if (socket_) {
      socket_->shutdown(ASIO_NS::ip::tcp::socket::shutdown_both);
      socket_->cancel();
      socket_->close();
    }
    io_service_->reset();
  }
  mutex_.unlock();
}

HTTPConnection::HTTPConnection(Application    &app,
                               RefHTTPRequest  request,
                               RefHTTPResponse response,
                               SlotComplete    slot_complete,
                               SlotError       slot_error,
                               SlotJoined      slot_joined) :
  Connection(app, request, response, slot_error, slot_joined),
  request_  (request),
  response_ (response)
{
  dispatcher_complete_.connect(sigc::bind(slot_complete, response_));
}
HTTPConnection::~HTTPConnection() {
  on_destroy();
}

#ifdef ENABLE_FILE_SCHEME
//========================
// FileRequest
FileRequest::FileRequest(const URL &url) :
  Request(url)
{
  if (strcasecmp(url.get_scheme().c_str(), "file") != 0) {
    throw Error::create(format(_("Invalid URL scheme - %s"),
                               url.get_scheme().c_str()));
  }
}

//========================
// FileConnection
void FileConnection::on_load() {
  mutex_.lock();
  cancellable_ = g_cancellable_new();
  mutex_.unlock();

  std::string url      = request_->get_url().to_string_base();
  GError     *err      = NULL;
  GFile      *file     = g_file_new_for_uri(url.c_str());
  gchar      *contents = NULL;
  gsize       length   = 0;
  if (g_file_load_contents(file, cancellable_,
                           &contents, &length,
                           NULL, &err)) {
    complete(contents, length);
  } else {
    error(err->message);
  }
  if (contents) g_free(contents);
  g_object_unref(file);
  if (err) g_error_free(err);

  mutex_.lock();
  g_object_unref(cancellable_);
  cancellable_ = NULL;
  mutex_.unlock();
}
void FileConnection::on_cancel() {
  Connection::on_cancel();
  mutex_.lock();
  if (cancellable_) {
    g_cancellable_cancel(cancellable_);
  }
  mutex_.unlock();
}

FileConnection::FileConnection(Application    &app,
                               RefFileRequest  request,
                               RefFileResponse response,
                               SlotComplete    slot_complete,
                               SlotError       slot_error,
                               SlotJoined      slot_joined) :
  Connection  (app, request, response, slot_error, slot_joined),
  cancellable_(NULL),
  request_    (request),
  response_   (response)
{
  dispatcher_complete_.connect(sigc::bind(slot_complete, response_));
}
FileConnection::~FileConnection() {
  on_destroy();
}
#endif

//========================
// Pool
std::auto_ptr<Pool> Pool::_instance;

void Pool::pre_http_request(RefHTTPRequest request) {
  ConfMain::RefNetwork nw = app_.conf().network();
  if (!request->get_has_auth() &&
      nw->peercast_server()->equal(request->get_url()) &&
      (!(*nw->peercast_server_id()).empty() ||
       !(*nw->peercast_server_pw()).empty())) {
    request->set_auth_basic(*nw->peercast_server_id(),
                            *nw->peercast_server_pw());
  }
}

void Pool::get_auth(const URL &url, std::string &id, std::string &pw) {
  if (url.has_auth()) {
    if (url.get_id()) id = *url.get_id();
    if (url.get_pw()) pw = *url.get_pw();
  } else {
    /*
    if (nw->peercast_server()->equal(url) &&
        (!(*nw->peercast_server_id()).empty() ||
         !(*nw->peercast_server_pw()).empty())) {
      id = *nw->peercast_server_id();
      pw = *nw->peercast_server_pw();
    }
    */
  }
}
RefHTTPRequest Pool::create_auth_request(RefRequest      request,
                                         RefHTTPResponse response,
                                         Glib::ustring  &error_message) {
  RefHTTPRequest http = boost::dynamic_pointer_cast<HTTPRequest>(request);
  if (http && !http->get_has_auth()) {
    std::string id;
    std::string pw;
    get_auth(request->get_url(), id, pw);
    if (!id.empty() || !pw.empty()) {
      const std::string &auth_scheme = response->get_auth_scheme();
      if (auth_scheme == "Basic") {
        http = http->copy();
        http->set_auth_basic(id, pw);
        return http;
      } else {
        error_message = format(_("Unsupported authorization - %s"),
                               auth_scheme.c_str()).c_str();
      }
    } else {
      error_message = _("Authorization required");
    }
  } else {
    error_message = _("Request is not HTTP");
  }
  return RefHTTPRequest();
}

void Pool::http_async_on_complete(const RefHTTPResponse response,
                                  AutoContext          *ac) {
  if (response->is_redirected()) {
    const std::string &url = response->get_redirect_location();
    try {
      RefHTTPRequest request = HTTPRequest::create(URL::build(ac->connection_->get_request()->get_url(), url));
      RefHTTPConnection conn = req(request, ac->slot_complete_, ac->slot_error_);
      ac->connection_->set_next(conn);
      log::dbg << "loader: redirected - " << url << log::endl;
      conn->set_timeout(ac->connection_->get_timeout());
      conn->load();
    } catch(const InvalidURL &ex) {
      RefError err = Error::create(format(_("Invalid redirection - %s"),
                                          url.c_str()));
      if (!ac->slot_error_.empty()) {
        ac->slot_error_(err);
      } else {
        log::err << "loader: " << err->get_message() << log::endl;
      }
    }
  } else if (response->is_auth_required()) {
    Glib::ustring err;
    RefHTTPRequest request =
      create_auth_request(ac->connection_->get_request(), response, err);
    if (request) {
      RefConnection conn = req(request, ac->slot_complete_, ac->slot_error_);
      ac->connection_->set_next(conn);
      conn->load();
    } else {
      if (!ac->slot_error_.empty()) {
        ac->slot_error_(Error::create(err));
      } else {
        log::err << "loader: " << err << log::endl;
      }
    }
  } else if (response->get_code() == 200) {
    if (!ac->slot_complete_.empty()) {
      ac->slot_complete_(response);
    } else {
      log::log << "loader: complete - " << ac->connection_->get_request()->get_url() << log::endl;
    }
  } else {
    RefError err = Error::create(format("%i %s",
                                        response->get_code(),
                                        response->get_message().c_str()));
    if (!ac->slot_error_.empty()) {
      ac->slot_error_(err);
    } else {
      log::err << "loader: " << err->get_message() << log::endl;
    }
  }
}
void Pool::conn_on_joined(AutoContext *ac) {
  for(ACList::iterator iter = auto_contexts_.begin(); iter != auto_contexts_.end(); ++iter) {
    if (&*iter == ac) {
      auto_contexts_.erase(iter);
      return;
    }
  }
  log::err << "bug: loader failed to find context - " << ac << log::endl;
}

RefHTTPConnection Pool::req(RefHTTPRequest request,
                            SlotComplete   slot_complete,
                            SlotError      slot_error) {
  using sigc::mem_fun;
  using sigc::bind;
  pre_http_request(request);
  log::dbg << request->get_url() << log::endl;
  AutoContext *ac = new AutoContext();
  RefHTTPConnection conn =
    HTTPConnection::create(app_,
                           request,
                           bind(mem_fun(*this, &Pool::http_async_on_complete), ac),
                           slot_error,
                           bind(mem_fun(*this, &Pool::conn_on_joined), ac));
  ac->connection_    = conn;
  ac->slot_complete_ = slot_complete;
  ac->slot_error_    = slot_error;
  auto_contexts_.push_back(ac);
  return conn;
}
#ifdef ENABLE_FILE_SCHEME
RefFileConnection Pool::req(RefFileRequest request,
                            SlotComplete   slot_complete,
                            SlotError      slot_error) {
  using sigc::mem_fun;
  using sigc::bind;
  log::dbg << request->get_url() << log::endl;
  AutoContext *ac = new AutoContext();
  RefFileConnection conn =
    FileConnection::create(app_,
                           request,
                           slot_complete,
                           slot_error,
                           bind(mem_fun(*this, &Pool::conn_on_joined), ac));
  ac->connection_    = conn;
  ac->slot_complete_ = slot_complete;
  ac->slot_error_    = slot_error;
  auto_contexts_.push_back(ac);
  return conn;
}
#endif
RefConnection Pool::req(const URL         &url,
                        SlotComplete       slot_complete,
                        SlotError          slot_error,
                        const std::string &charset) {
  const std::string &scheme = url.get_scheme();
  if (strcasecmp(scheme.c_str(), "http") == 0) {
    RefHTTPRequest request = HTTPRequest::create(url);
    request->set_charset(charset);
    return req(request, slot_complete, slot_error);
#ifdef ENABLE_FILE_SCHEME
  } else if (strcasecmp(scheme.c_str(), "file") == 0) {
    RefFileRequest request = FileRequest::create(url);
    request->set_charset(charset);
    return req(request, slot_complete, slot_error);
#endif
  } else {
    Glib::ustring msg = format(_("Unknown URL sheme - %s"), scheme.c_str());
    if (!slot_error.empty()) {
      slot_error(Error::create(msg));
    } else {
      log::err << msg << log::endl;
    }
    return RefConnection();
  }
}

RefHTTPResponse Pool::req_sync(RefHTTPRequest request) {
  using sigc::mem_fun;
  using sigc::bind;
  RefHTTPConnection conn_prev;
  while(1) {
    /* req()ではcompleteコールバックに専用のプロクシを挟むので再利用しない。 */
    log::dbg << request->get_url() << log::endl;
    pre_http_request(request);
    AutoContext *ac = new AutoContext();
    RefHTTPConnection conn =
      HTTPConnection::create(app_,
                             request,
                             HTTPConnection::SlotComplete(),
                             HTTPConnection::SlotError(),
                             bind(mem_fun(*this, &Pool::conn_on_joined), ac));
    ac->connection_ = conn;
    auto_contexts_.push_back(ac);
    if (conn_prev) conn_prev->set_next(conn);
    conn_prev = conn;
    RefHTTPResponse response = conn->load_sync();
    if (response->is_redirected()) {
      const std::string &url = response->get_redirect_location();
      try {
        request = HTTPRequest::create(URL::build(request->get_url(), url));
      } catch(const InvalidURL &ex) {
        throw Error::create(format(_("Invalid redirection - %s"),
                                   url.c_str()));
      }
    } else if (response->is_auth_required()) {
      Glib::ustring err;
      request = create_auth_request(request, response, err);
      if (!request) throw Error::create(err);
    } else if (response->get_code() == 200) {
      return response;
    } else {
      throw Error::create(format("%i %s",
                                 response->get_code(),
                                 response->get_message().c_str()));
    }
  }
  return RefHTTPResponse();
}
#ifdef ENABLE_FILE_SCHEME
RefFileResponse Pool::req_sync(RefFileRequest request) {
  return req(request, SlotComplete(), SlotError())->load_sync();
}
#endif
RefResponse Pool::req_sync(const URL         &url,
                           const std::string &charset) {
  const std::string &scheme = url.get_scheme();
  if (strcasecmp(scheme.c_str(), "http") == 0) {
    RefHTTPRequest request = HTTPRequest::create(url);
    request->set_charset(charset);
    return req_sync(request);
#ifdef ENABLE_FILE_SCHEME
  } else if (strcasecmp(scheme.c_str(), "file") == 0) {
    RefFileRequest request = FileRequest::create(url);
    request->set_charset(charset);
    return req_sync(request);
#endif
  } else {
    throw Error::create(format(_("Unknown URL sheme - %s"), scheme.c_str()));
  }
}

Pool::Pool(Application &app) :
  app_(app)
{
}
Pool::~Pool() {
}

}/*loader*/
}/*gpyp*/
